package com.mango.ch08;

import java.io.IOException;
import java.util.TreeSet;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

/**
 * 一下Hadoop内置数据类型 只能作为values使用
 * ArrayWritable：存储属于Writable类型的值数组。要使用ArrayWritable类型作为reduce输入的value类型，则需要创建ArrayWritable的子类来指定存储在其中的Writable值的类型。
 * TwoDArrayWritable：存储属于同一Writable类型的值矩阵。要使用TwoDArrayWritable类型作为reduce输入的value类型，则需要通过创建与ArrayWritable类型相似的TwoDArrayWritable类型的子类来指定存储的值的类型。
 * MapWritable：存储键值对的映射。键和值应该是Writable数据类型。
 * SortedMapWritable：存储键值对的有序映射。键应该实现writableComparable接口。
 * 
 * @author Mango
 *
 */
public class ComFriendsMapper extends Mapper<LongWritable, Text, Text, Text> {

	@Override
	protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
		// TODO Auto-generated method stub

		String[] takens = value.toString().split(",");
		String user = takens[0];
		String[] friends = takens[1].split(" ");
		System.out.println("takens" + takens);
		TreeSet<Integer> rightTreeSet = getFriend(friends);
		System.out.println("输出friends 列表" + rightTreeSet);
		for (String f : friends) {
			TreeSet<Integer> leftTreeSet = new TreeSet<>();
			leftTreeSet.add(Integer.valueOf(f));
			leftTreeSet.add(Integer.valueOf(user));
			System.out.println("user:" + leftTreeSet + " friends:" + rightTreeSet);
			context.write(new Text(leftTreeSet.toString()), new Text(rightTreeSet.toString()));
		}

	}

	public TreeSet<Integer> getFriend(String[] fris) {
		TreeSet<Integer> TreeSet = new TreeSet<>();
		for (String string : fris) {
			TreeSet.add(Integer.valueOf(string));
		}
		System.out.println("返回的TreeSet集合的值是" + TreeSet);
		return TreeSet;
	}
}
